package com.shujia.utils;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.regionserver.BloomType;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.List;

/**
 *  工具类的编写步骤：
 *  1、构造方法私有化
 *  2、方法要是静态的，将可以通过类名的方式调用
 *
 */
public class HbaseUtil {
    //成员变量
    public static final Connection CONN = HbaseParameters.getConnection();
    public static final Admin ADMIN = HbaseParameters.getAdmin();

    private HbaseUtil(){}

    /**
     * 传入表名和若干个列簇创建Hbase一张表
     */
    public static void createOneTable(String tableName, String ... familyNames){
        TableName name = TableName.valueOf(tableName);

        TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(name);

        try {
            for (String familyName : familyNames) {
//                ColumnFamilyDescriptor info = ColumnFamilyDescriptorBuilder.of(familyName);
                ColumnFamilyDescriptor info = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(familyName))
                        .setBloomFilterType(BloomType.ROW) // 设置布隆过滤器
                        .build();

                tableDescriptorBuilder.setColumnFamily(info);
            }

            if (ADMIN.tableExists(name)) {
                System.out.println(Bytes.toString(name.getName()) + " 表已经存在！");
                return;
            }
            ADMIN.createTable(tableDescriptorBuilder.build());
            System.out.println(Bytes.toString(tableDescriptorBuilder.build().getTableName().getName()) + "表创建 成功 SUCCEED！");
        } catch (Exception e) {
            System.out.println(Bytes.toString(tableDescriptorBuilder.build().getTableName().getName()) + "表创建 失败！FAILED！");
            e.printStackTrace();
        }
    }

    /**
     * 删除表
     */
    public static void dropTable(String tableName){
        try {
            //将表名封装成一个TableName对象
            TableName name = TableName.valueOf(tableName);

            /**
             * 判断表是否存在
             */
            if (!ADMIN.tableExists(name)) {
                System.out.println(Bytes.toString(name.getName()) + " 表不存在！无法删除！");
                return;
            }


            //禁用表
            ADMIN.disableTable(name);

            //删除表
            ADMIN.deleteTable(name);

            System.out.println(Bytes.toString(name.getName()) + "表删除 成功 SUCCEED！");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 清空表数据
     */
    public static void truncateTable(String tableName){
        try {
            //将表名封装成一个TableName对象
            TableName name = TableName.valueOf(tableName);

            /**
             * 判断表是否存在
             */
            if (!ADMIN.tableExists(name)) {
                System.out.println(Bytes.toString(name.getName()) + " 表不存在！无法删除！");
                return;
            }

            //注意：如果是使用程序清空数据，需要手动设置禁用
            ADMIN.disableTable(name);

            ADMIN.truncateTable(name,true);

            System.out.println(Bytes.toString(name.getName()) + "表数据清除 成功 SUCCEED！");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 获取一张表
     */
    public static Table getOneTable(String tableName){
        Table table = null;

        try {
            //将表名封装成一个TableName对象
            TableName name = TableName.valueOf(tableName);

            /**
             * 判断表是否存在
             */
            if (!ADMIN.tableExists(name)) {
                System.out.println(Bytes.toString(name.getName()) + " 表不存在！");
            }

            //获取表实例对象
            table = CONN.getTable(name);
        }catch (Exception e){
            e.printStackTrace();
        }
        return table;
    }

    /**
     * 添加一列数据
     */
    public static void putOneColData(String tableName, String rowKey, String familyName, String colName, String colValue){
        try {
            //将表名封装成一个TableName对象
            TableName name = TableName.valueOf(tableName);

            /**
             * 判断表是否存在
             */
            if (!ADMIN.tableExists(name)) {
                System.out.println(Bytes.toString(name.getName()) + " 表不存在！无法添加数据！");
                return;
            }

            //获取表实例对象
            Table table = CONN.getTable(name);

            //将一列数据封装成一个Put对象,传入一个行键，需要将行键变成一个字节数组的格式
            Put put = new Put(Bytes.toBytes(rowKey));
            KeyValue keyValue = new KeyValue(Bytes.toBytes(rowKey),
                    Bytes.toBytes(familyName),
                    Bytes.toBytes(colName),
                    Bytes.toBytes(colValue));
            put.add(keyValue);

            table.put(put);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    /**
     * 打印一行多列数据
     */
    public static void printResult(Result result){
        String rk = Bytes.toString(result.getRow());
        List<Cell> cells = result.listCells();
        StringBuilder sb = new StringBuilder();
        sb.append("行键：").append(rk).append(", ");
        for (int i = 0; i < cells.size(); i++) {
            //Hbase中提供了一个专门解析Cell单元格的工具类 CellUtil
            String colName = Bytes.toString(CellUtil.cloneQualifier(cells.get(i)));
            String colValue = Bytes.toString(CellUtil.cloneValue(cells.get(i)));
            if (i != cells.size() - 1) {
                sb.append(colName).append("：").append(colValue).append(", ");
            } else {
                sb.append(colName).append("：").append(colValue);
            }
        }
        System.out.println(sb);
        System.out.println("--------------------------------------------");
    }
}
